package com.hugh.hdfs.api;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

/**
 * @Author: Hugh
 * @Date: 2020/9/9
 */
public class HdfsClientTest {

    private FileSystem fs;

    private Configuration configuration;

    @Before
    public void initFs() throws URISyntaxException, IOException, InterruptedException {
        // 1 获取文件系统
        configuration = new Configuration();
        configuration.set("dfs.replication", "2");
        // 配置在集群上运行
        // configuration.set("fs.defaultFS", "hdfs://linux121:9000");
        // FileSystem fs = FileSystem.get(configuration);
        fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root");
    }

    @After
    public void clearFs() throws IOException {
        // 3 关闭资源
        fs.close();
    }

    @Test
    public void testMkdirs() throws IOException, InterruptedException, URISyntaxException {

        // 2 创建目录
        fs.mkdirs(new Path("/guleHaha"));
    }

    /**
     * 上传文件
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void testCopyFromLocalFile() throws IOException, InterruptedException, URISyntaxException {

        // 2 上传文件
        //通过配置文件
        /**
         * 参数优先级
         *
         * 参数优先级排序：（
         * 1）代码中设置的值 >（2）用户自定义配置文件 >（3）服务器的默认配置
         */
        fs.copyFromLocalFile(new Path("/Users/gule/java/HughGl/HughGl-stage4-5-hadoop-hbase/README.md"), new Path("/guleHaha/README.md"));
    }

    /**
     * 下载文件
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void testCopyToLocalFile() throws IOException, InterruptedException, URISyntaxException{

        // 2 执行下载操作
        // boolean delSrc 指是否将原文件删除
        // Path src 指要下载的文件路径
        // Path dst 指将文件下载到的路径
        // boolean useRawLocalFileSystem 是否开启文件校验
        fs.copyToLocalFile(false, new Path("/guleHaha/README.md"), new Path("README.txt"), true);
    }

    /**
     * 删除文件/文件夹
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void testDelete() throws IOException, InterruptedException, URISyntaxException{
        // 2 创建目录
        fs.mkdirs(new Path("/api_test"));
        fs.copyFromLocalFile(new Path("/Users/gule/java/HughGl/HughGl-stage4-5-hadoop-hbase/README.md"), new Path("/api_test/README.md"));
        // 2 执行删除
        fs.delete(new Path("/api_test/"), true);
    }

    /**
     * 查看文件名称、权限、长度、块信息
     */
    @Test
    public void testListFiles() throws IOException, InterruptedException, URISyntaxException{
        // 2 获取文件详情
        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
        while(listFiles.hasNext()) {
            LocatedFileStatus status = listFiles.next();
            // 输出详情
            // 文件名称
            System.out.println(status.getPath().getName());
            // 长度
            System.out.println(status.getLen());
            // 权限
            System.out.println(status.getPermission());
            // 分组
            System.out.println(status.getGroup());
            // 获取存储的块信息
            BlockLocation[] blockLocations = status.getBlockLocations();
            for (BlockLocation blockLocation : blockLocations) {
                // 获取块存储的主机节点
                String[] hosts = blockLocation.getHosts();
                for (String host : hosts) { System.out.println(host); }
            }
        }
        System.out.println("-----------华丽的分割线----------");
    }

    /**
     * 文件夹判断
     */
    @Test
    public void testListStatus() throws IOException, InterruptedException, URISyntaxException{
        // 2 判断是文件还是文件夹
        FileStatus[] listStatus = fs.listStatus(new Path("/"));
        for (FileStatus fileStatus : listStatus) {
            // 如果是文件
            if (fileStatus.isFile()) {
                System.out.println("f:"+fileStatus.getPath().getName());
            }else {
                System.out.println("d:"+fileStatus.getPath().getName());
            }

        }
        System.out.println("-----------华丽的分割线----------");
    }

    /**
     * I/O流操作HDFS之文件上传
     */
    @Test
    public void putFileToHDFS() throws IOException, InterruptedException, URISyntaxException {
        //1. 读取本地文件的输入流
        final FileInputStream inputStream = new FileInputStream(new File("/Users/gule/java/HughGl/HughGl-stage4-5-hadoop-hbase/README.md"));
        //2. 准备写数据到hdfs的输出流
        final FSDataOutputStream outputStream = fs.create(new Path("/hehe/halou.txt"));
        // 3.输入流数据拷贝到输出流 :数组的大小，以及是否关闭流底层有默认值
        IOUtils.copyBytes(inputStream, outputStream, configuration);
        // 4.可以再次关闭流
        IOUtils.closeStream(outputStream);
        IOUtils.closeStream(inputStream);
        System.out.println("-----------华丽的分割线----------");
    }

    /**
     * I/O流操作HDFS之文件下载
     */
    @Test
    public void downLoadFileToHDFS() throws IOException, InterruptedException, URISyntaxException {
        //1. 读取hdfs文件的输入流
        final FSDataInputStream in = fs.open(new Path("/hehe/halou.txt"));
        //2. 本地文件的输出流
        final FileOutputStream out = new FileOutputStream(new File("halou"));
        // 3.输入流数据拷贝到输出流 :数组的大小，以及是否关闭流底层有默认值
        IOUtils.copyBytes(in, out, configuration);
        // 4.可以再次关闭流
        IOUtils.closeStream(out);
        IOUtils.closeStream(in);
        System.out.println("-----------华丽的分割线----------");
    }

    //seek定位读取hdfs指定文件 :
    // 使用io流读取/lagou.txt文件并把内容输出两次，本质就是读取文 件内容两次并输出
    @Test public void seekReadFile() throws IOException {

        //1 创建一个读取hdfs文件的输入流
        final FSDataInputStream in = fs.open(new Path("/hehe/halou.txt"));
        //2.控制台数据：System.out
        //3 实现流拷贝，输入流--》控制台输出
        // IOUtils.copyBytes(in, System.out, configuration);
        IOUtils.copyBytes(in, System.out, 4096, false);
        //4. 再次读取文件
        in.seek(0);
        // 定位从0偏移量（文件头部）再次读取
        IOUtils.copyBytes(in, System.out, 4096, false);

        //5.关闭输入流
        IOUtils.closeStream(in);
        System.out.println("-----------华丽的分割线----------");
    }

    /**
     * 验证Packet代码
     * @throws IOException
     */
    @Test
    public void testUploadPacket() throws IOException {
        //1 准备读取本地文件的输入流
        final FileInputStream in = new FileInputStream(new File("/Users/gule/java/HughGl/HughGl-stage4-5-hadoop-hbase/README.md"));

        //2 准备好写出数据到hdfs的输出流
        final FSDataOutputStream out = fs.create(new Path("/hehe/halou2.txt"), new Progressable() {
            @Override
            public void progress(){
                //这个progress方法就是每传输64KB（packet）就会执 行一次，
            System.out.println("&"); }
        });
        //3 实现流拷贝
        IOUtils.copyBytes(in, out, configuration); //默认关闭流选项是true，所以会自动关闭
        //4 关流可以再次关闭也可以不关了
    }
}
